package com.duoec.dw.lineage.core.utils;

import com.duoec.dw.lineage.exceptions.LineageAnalyseException;
import com.duoec.dw.lineage.dto.TableFieldIndex;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.CSVRecord;
import org.junit.Test;
import org.springframework.util.CollectionUtils;

import java.io.*;
import java.util.Collections;
import java.util.List;
import java.util.Map;

public class TableFieldSplitter {
    private static Map<String, List<TableFieldIndex>> tableMap = Maps.newHashMap();
    private static final String CSV_DIR = "/Users/xuwenzhen/Projects/duoec/d/duo-lineage/lineage-druid/src/test/resources/table";
    private static final String[] FILE_HEADER = {"col_name", "data_type", "comment"};

    @Test
    public void splitter() throws Exception {
        Reader in = new FileReader("/Users/xuwenzhen/Downloads/table_field_comment.csv");
        Iterable<CSVRecord> records = CSVFormat.EXCEL.withQuote('"').parse(in);
        for (CSVRecord record : records) {
            String dbName = record.get(0);
            String tableName = record.get(1);
            if (dbName.equals("DB_NAME") && tableName.equals("TBL_NAME")) {
                continue;
            }
            String columnName = record.get(2);
            String columnType = record.get(3);
            String columnComment = record.get(4);
            String index = record.get(5);
            readLine(Integer.parseInt(index), dbName, tableName, columnName, columnComment, columnType);
        }

        tableMap.forEach((key, tableFieldIndexList) -> {
            try {
                String[] ks = key.split("\\.");
                File dir = new File(CSV_DIR, ks[0]);
                if (!dir.exists()) {
                    dir.mkdir();
                }
                if (CollectionUtils.isEmpty(tableFieldIndexList)) {
                    return;
                }
                Collections.sort(tableFieldIndexList);

                writeCsv(tableFieldIndexList, new File(dir, ks[1] + ".csv"));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    private void writeCsv(List<TableFieldIndex> tableFieldIndexList, File to) throws IOException {
        CSVFormat csvFormat = CSVFormat.DEFAULT.withHeader(FILE_HEADER);
        try (FileWriter fileWriter = new FileWriter(to); CSVPrinter csvPrinter = new CSVPrinter(fileWriter, csvFormat)) {
            tableFieldIndexList.forEach(tfi -> {
                List<String> record = Lists.newArrayList();
                record.add(tfi.getColumnName());
                record.add(tfi.getColumnType());
                record.add(tfi.getColumnComment());
                try {
                    csvPrinter.printRecord(record);
                } catch (IOException e) {
                    throw new LineageAnalyseException("写入csv失败！", e);
                }
            });
        }
    }

    private void readLine(Integer index, String dbName, String tableName, String columnName, String columnComment, String columnType) {
        String key = dbName + "." + tableName;

        List<TableFieldIndex> fields = tableMap.computeIfAbsent(key, k -> Lists.newArrayList());
        TableFieldIndex tfi = new TableFieldIndex();
        tfi.setColumnComment(columnComment);
        tfi.setColumnName(columnName);
        tfi.setColumnType(columnType);
        tfi.setIndex(index);
        fields.add(tfi);
    }

    @Test
    public void setTableCsv() {
        File csvDir = new File(CSV_DIR);
        scanDir(csvDir);
    }

    private void scanDir(File csvDir) {
        File[] fs = csvDir.listFiles();
        for (File file : fs) {
            if (file.isDirectory()) {
//                scanDir(file);
                continue;
            }
            String name = file.getName();

            if (name.endsWith(".csv")) {
                String[] ks = name.split("\\.");
                File dir = new File(csvDir, ks[0]);
                if (!dir.exists()) {
                    dir.mkdir();
                }
                file.renameTo(new File(dir, ks[1] + ".csv"));
            }
        }
    }
}
